#include"rpcprovider.h"
#include"mprpc.h"
#include"zookeeperuti.h"

//发布rpc方法
void RpcProvider::NotifyServic(google::protobuf::Service* service)
{
    ServerInfo info;

    //服务对象的描述信息
    const google::protobuf::ServiceDescriptor *Desc=service->GetDescriptor();
    //获取服务信息
    std::string service_name=Desc->name();
    //获取服务对象的方法数量
    int service_cnt=Desc->method_count();


    for(int i=0;i<service_cnt;i++)
    {
        //获取服务方法的描述
        const google::protobuf::MethodDescriptor *method_desc=Desc->method(i);
        std::string method_name=method_desc->name();
        info._method_map.insert({method_name,method_desc});
    }
    info._server=service;
    _serverMap.insert({service_name,info});
}


//启动rpc服务
void RpcProvider::Run()
{
    std::string ip = Mprpc::GetConfig().GetItem("rpcserverip");
    uint16_t port = atoi(Mprpc::GetConfig().GetItem("rpcserverport").c_str());

    //注册所有服务
    ZkClient zkcli;
    zkcli.Start();
    for(auto& sp:_serverMap)
    {
        std::string service_path="/"+sp.first;
        zkcli.Create(service_path.c_str(),nullptr,0);
        for(auto& mp:sp.second._method_map)
        {
            std::string method_path=service_path+"/"+mp.first;
            char method_path_data[128]={0};
            sprintf(method_path_data,"%s:%d",ip.c_str(),port);
            //创建临时结点
            zkcli.Create(method_path.c_str(),method_path_data,strlen(method_path_data),ZOO_EPHEMERAL);
        }
    }

    muduo::net::InetAddress add(ip,port);
    
    //创建TcpServer
    muduo::net::TcpServer server(&_epoll,add,"RpcProvider");

    //绑定连接回调  通过bind将类内函数转换为普通函数
    server.setConnectionCallback(std::bind(&RpcProvider::Connection,this,std::placeholders::_1));

    //绑定读写回调
    server.setMessageCallback(std::bind(&RpcProvider::Message,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));


    //设置线程数
    int rpcthreadnum=atoi(Mprpc::GetConfig().GetItem("rpcthreadnum").c_str());
    server.setThreadNum(4);

    //启动网络服务
    server.start();
    _epoll.loop();
}


//连接回调
void RpcProvider::Connection(const muduo::net::TcpConnectionPtr& con)
{
    //连接断开
    if(!con->connected())
    {
        con->shutdown();
    }
}


//读写回调
void RpcProvider::Message(const muduo::net::TcpConnectionPtr& con,muduo::net::Buffer* buf,muduo::Timestamp receiveTime)
{
    //读4字节 报头长度
    std::string recv_buf(buf->retrieveAllAsString());
    int recv_size=recv_buf.size();
    uint32_t header_size;
    recv_buf.copy((char*)&header_size,4,0);

    mprpc::RpcHeader rpcheader;

    std::string header(recv_buf.substr(4,header_size));
    //反序列化
    if(!rpcheader.ParseFromString(header))
    {
        std::cout << "rpcheader parse false!" << std::endl; 
        return;
    }

    //对象名
    std::string service_name=rpcheader.service_name();
    //方法名
    std::string method_name=rpcheader.method_name();
    //参数长度
    int args_size=rpcheader.args_size();

    //读参数
    std::string args(recv_buf.substr(4+header_size,16));


    //找到方法描述
    auto it=_serverMap.find(service_name);
    if(it==_serverMap.end())
    {
        std::cout << "service not find!" << std::endl; 
        return;
    }
    
    google::protobuf::Service* service=it->second._server;
    
    auto mit=it->second._method_map.find(method_name);
    if(mit==it->second._method_map.end())
    {
        std::cout << "Method not find!" << std::endl; 
        return;
    }
    const google::protobuf::MethodDescriptor* method_desc=mit->second;

    //调用方法
    google::protobuf::Message *request=service->GetRequestPrototype(method_desc).New();
    if(!request->ParseFromString(args))
    {
        std::cout << "request parse false!" << std::endl; 
        return;
    }
    google::protobuf::Message* response=service->GetResponsePrototype(method_desc).New();

    //创建回调函数
    google::protobuf::Closure* done=google::protobuf::NewCallback<RpcProvider,
                                                                  const muduo::net::TcpConnectionPtr&,
                                                                  google::protobuf::Message*   
                                                                  >(this,&RpcProvider::sendRpcResponse,con,response);

    //执行rpc方法
    service->CallMethod(method_desc,nullptr,request,response,done);
}



//发送响应
void RpcProvider::sendRpcResponse(const muduo::net::TcpConnectionPtr& con,google::protobuf::Message* Response)
{
    std::string response_str;
    if(!Response->SerializeToString(&response_str))
    {
        std::cout << "Response SerializeAsString false!" << std::endl; 
        con->shutdown();
        return;
    }
    con->send(response_str);
    con->shutdown();
}



